package com.wyg.epkafkastarter.producer;

import com.wyg.epkafkastarter.constant.Topic;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

/**
 * @Author: WangYongGan
 * @Date: 2022/5/17 16:32
 * @Description:
 */
@Component
public class BaseProducer {

    public static KafkaTemplate<String, String> kafkaTemplate;

    public static void send(Topic topic, String msg) {
        final ListenableFuture<SendResult<String, String>> res = kafkaTemplate.send(topic.getValue(), msg);
    }


    public static void send(Topic topic, String msg, SuccessCallback<? super SendResult<String, String>> success, FailureCallback error) {
        final ListenableFuture<SendResult<String, String>> resFuture = kafkaTemplate.send(topic.getValue(), msg);
        resFuture.addCallback(success, error);
    }

}
